with t_collect_seven_base as(
select
    billcode
    ,manage_region_code
    ,manage_region_name
    ,collect_agent_code
    ,collect_agent_name
    ,center_scansitecode
    ,center_inputsite
    ,collect_scansitecode
    ,collect_inputsite
    ,dispatch_network_code
    ,dispatch_network_name
    ,order_source --订单来源
    ,center_scantime --中心实际发件时间
    ,collect_planned_dep_date   --集散规划发件日期
    ,collect_planned_departure_time
    ,center_planned_dep_date  --中心规划发件日期
    ,center_planned_departure_time
    ,center_planned_arr_date  --中心规划到达日期
    ,center_planned_arrival_time
    ,if(arrival_scan_time is not null, 1, 0) as is_collect_arrival ---集散是否到件，1到件
    ,arrival_scan_time as collect_arrival_scan_time  --集散到件日期时间
    ,arrival_scan_inputsite --集散到件网点
    ,send_scan_time     --集散发件日期时间
    ,send_scan_inputsite    --集散发件网点
    ,collect_send_in_time   --散货及时标识，计算散货及时票数
    ,is_dispatch_network
    ,is_entrepot_exception
    ,send_date
    ,dt
from jms_dm.dm_collect_analysis_detail
where dt between date_add('{{ execution_date  | cst_ds }}',-6) and date_add('{{ execution_date  | cst_ds }}',0)
and (collect_send_in_time=1 or collect_send_in_time=0 or is_dispatch_network=1 or is_entrepot_exception =1)
),
--任务单下一站为集散点的数据
t_center_shipment as(
 select centersend.scantime,
        centersend.billcode ,
        centersend.inputsite,
        centersend.scansitecode,
        centersend.nextstation,
        centersend.nextstationcode,
        centersend.transfercode,
        date_format(centersend.scantime, 'hh:mm:ss') scantime_hm,
        net.agent_name as collect_agent_name,
        net.agent_code as collect_agent_code
 from (select waybill_no as billcode
            , scan_time as scantime
            , scan_site as inputsite
            , scan_site_code as scansitecode
            , next_station as nextstation
            , next_station_code as nextstationcode
            , transfer_code as transfercode
            ,  row_number() over (partition by waybill_no, scan_site_code,next_station_code order by scan_time desc) bill_rank
       from jms_dwd.dwd_tab_barscan_centersend_base_dt
       where dt between date_add('{{ execution_date  | cst_ds }}',-12) and date_add('{{ execution_date  | cst_ds }}',0)
         and next_station_code is not null
         and next_network_type=3
       )centersend left join jms_dim.dim_network_whole_massage net on centersend.nextstationcode=net.code
       where bill_rank = 1
),
--集散发件单号【全量】，集散点和单号去重
t_collect_send_bill_unrepeat as(
select
    billcode
    ,scansitecode
    ,scantime
from (
    select
         waybill_no as billcode
        ,scan_site_code as scansitecode
        ,scan_time as scantime
        ,row_number() over (partition by scan_site_code,waybill_no order by scan_time desc) rnk --集散发件取最早的，不然发来发去不好比较
    from jms_dwd.dwd_tab_barscan_sitesend_base_dt
    where dt between date_add('{{ execution_date  | cst_ds }}',-9) and date_add('{{ execution_date  | cst_ds }}',0)
    and network_type=3
)t_collect_send_bill where rnk = 1
),
t_network_site as(
select * from(
  SELECT
  code
  ,name
  ,type_id
  ,fran_code AS franchisee_code
  ,fran_name AS franchisee_name
  ,row_number() over (partition by code order by update_time desc) rnk
  FROM jms_dim.dim_network_whole_massage
  where type_id = '336' and is_entrepot = 2 and is_enable=1 and is_delete = 1
)tmp where rnk = 1
),
--只是集散点
t_network_collect as(
select * from(
  SELECT
  code
  ,name
  ,type_id
  ,fran_code AS franchisee_code
  ,fran_name AS franchisee_name
  ,row_number() over (partition by code order by update_time desc) rnk
  FROM jms_dim.dim_network_whole_massage
  where type_id = '336' and is_entrepot = 1 and is_enable=1 and is_delete = 1
)tmp where rnk = 1
),
--网点到发签单号【全量】,单号去重取最晚扫描,网点叫做派件扫描
t_network_arrival_send_sign_bill_unrepeat as(
SELECT
billcode,scantime,scansitecode
from (
select
*,
row_number() over (partition by billcode order by scantime desc) rnk
from (
    select
        waybill_no as billcode
        ,scan_time as scantime
        ,scan_site_code as scansitecode
    from jms_dwd.dwd_tab_barscan_sitearrival_base_dt t1
    where dt between date_add('{{ execution_date  | cst_ds }}',-9) and date_add('{{ execution_date  | cst_ds }}',0)
             and network_type=4
    union all
    select
        waybill_no as billcode
        ,scan_time as scantime
        ,scan_site_code as scansitecode
    from jms_dwd.dwd_tab_barscan_sitesend_base_dt  t1
    where dt between date_add('{{ execution_date  | cst_ds }}',-9) and date_add('{{ execution_date  | cst_ds }}',0)
          and network_type=4
    union all
    select
       waybill_no as billcode
      ,signed_time as scantime --签收扫描时间是这个字段
      ,signed_site_code as scansitecode
    from jms_dwd.dwd_tab_barscan_sign_base_dt where dt between date_add('{{ execution_date  | cst_ds }}',-9) and date_add('{{ execution_date  | cst_ds }}',0)
    union all
    select
         waybill_no as billcode
        ,scan_time as scantime
        ,scan_site_code as scansitecode
    from jms_dwd.dwd_tab_barscan_collect_base_dt
    where dt between date_add('{{ execution_date  | cst_ds }}',-9) and date_add('{{ execution_date  | cst_ds }}',0)
          and (scan_type='出库扫描' or scan_type = '快件取出扫描')
    union all
    select
         waybill_no as billcode
        ,scan_time as scantime
        ,scan_site_code as scansitecode
    from jms_dwd.dwd_tab_barscan_deliver_base_dt where dt between date_add('{{ execution_date  | cst_ds }}',-9) and date_add('{{ execution_date  | cst_ds }}',0)

  )a
) tmp1 where rnk=1
),
---集散和中心到件和发件【全量】
t_collect_center_arrival_send_bill as(
select
    waybill_no as billcode
    ,scan_time as scantime
    ,scan_site_code as scansitecode
from jms_dwd.dwd_tab_barscan_arrival_base_dt where dt between date_add('{{ execution_date  | cst_ds }}',-9) and date_add('{{ execution_date  | cst_ds }}',0)
union all
select
    waybill_no as billcode
    ,scan_time as scantime
    ,scan_site_code as scansitecode
from jms_dwd.dwd_tab_barscan_send_base_dt where dt between date_add('{{ execution_date  | cst_ds }}',-9) and date_add('{{ execution_date  | cst_ds }}',0)
),
---集散和中心到件和发件【全量】【去重】
t_collect_center_arrival_send_bill_unrepeat as(
select * from (
  select
    *,
    row_number() over (partition by billcode order by scantime desc) rnk
  from(
    select
      t1.*
    from t_collect_center_arrival_send_bill t1
    left anti join t_network_site t2      ---剔除网点的卸车和到件扫描
      on t1.scansitecode = t2.code
    left anti join t_collect_seven_base t3    --剔除base表中集散的那些单号,集散到件和发件都是不是null
      on t1.billcode = t3.billcode and t1.scansitecode = t3.collect_scansitecode
      and (t3.send_scan_time is not null or t3.collect_arrival_scan_time is not null)
  ) tmp1
) tmp2 where rnk=1
),
--关联任务单和加盟商
t_union_1 as(
select
t1.*
,t2.transfercode as shipment_no --中心发集散任务单号
,t3.fran_code as franchisee_code --集散对应加盟商
,t3.fran_name as franchisee_name
from t_collect_seven_base t1
left join t_center_shipment t2
  on  t1.billcode = t2.billcode
  and t1.center_scansitecode = t2.scansitecode
    and t1.collect_scansitecode = t2.nextstationcode
left join jms_dim.dim_network_whole_massage t3
  on t1.collect_scansitecode = t3.code
),
--计算已散货标识符
t_union_2 as(
select
 t1.billcode
,t1.shipment_no
,t1.manage_region_code
,t1.manage_region_name
,t1.collect_agent_code
,t1.collect_agent_name
,t1.franchisee_code
,t1.franchisee_name
,t1.center_scansitecode
,t1.center_inputsite
,t1.collect_scansitecode
,t1.collect_inputsite
,t1.dispatch_network_code
,t1.dispatch_network_name
,t1.order_source --订单来源
,t1.collect_planned_dep_date   --集散规划发件日期
,t1.collect_planned_departure_time
,t1.center_planned_dep_date --中心规划发件日期
,t1.center_planned_departure_time
,t1.center_planned_arr_date --中心规划到达日期
,t1.center_planned_arrival_time
,t1.center_scantime --中心实际发件日期时间
,t1.collect_arrival_scan_time  --集散到件日期时间
,t1.arrival_scan_inputsite --集散到件网点
,t1.send_scan_time      --集散发件日期时间
,t1.send_scan_inputsite    --集散发件网点
,t2.scantime as scantime2  --后面的日期该集散有木有有做发件扫描时间
,t3.scantime as scantime3  --网点扫描时间
,t1.is_dispatch_network
,t1.is_entrepot_exception
,t1.collect_send_in_time   --散货及时标识，计算散货及时票数
,case
  when collect_send_in_time = 1 then 1
  when t2.scantime is not null then 1 --集散点后面做了发件扫描也是1
  when t3.scantime is not null then 1 --网点做了到派签扫描也是1
  else 0
end is_already_flag  --已散货票数标识
,case
  when collect_send_in_time = 0 and t2.scantime is null and t3.scantime is null and t4.scantime is null then 1
  when collect_send_in_time = 0 and t1.collect_planned_departure_time is null and t2.scantime is null and t3.scantime is null then 1
  else 0
end is_ready_flag  --待散货票数标识
,case
  when collect_send_in_time = 0 and t1.collect_planned_departure_time is not null and t2.scantime is null and t4.scantime is not null then 1 --当前集散点没有发件但是其它集散有做到发
  else 0
end is_wrong_flag  --错分票标识
,t1.is_collect_arrival ---集散是否到件，1到件
,t1.send_date
,t1.dt
from t_union_1 t1
left join t_collect_send_bill_unrepeat t2 --看再后面的日期该集散有木有有做发件扫描
  on t1.collect_scansitecode = t2.scansitecode and t1.billcode = t2.billcode and t2.scantime>t1.center_scantime
left join t_network_arrival_send_sign_bill_unrepeat t3 --网点的扫描时间要再集散到件时间或者发件时间之后
    on t1.billcode = t3.billcode and t3.scantime > coalesce(t1.collect_arrival_scan_time,t2.scantime,t1.center_scantime)
left join t_collect_center_arrival_send_bill_unrepeat t4 --判断其它集散和中心是否有到发件,不等于会发散
  on t1.billcode = t4.billcode and t4.scantime > coalesce(t1.collect_arrival_scan_time,t2.scantime,t1.center_scantime)
),
tab_is_reback as (
select
waybill_no
from
jms_dwd.dwd_tab_reback_transfer_express_base
where date(apply_time)>=date_add('{{ execution_date  | cst_ds }}',-30) and  date(apply_time)<=date_add('{{ execution_date  | cst_ds }}',0)
group by waybill_no
),
t_final_res as(
select
billcode
,shipment_no
,manage_region_code
,manage_region_name
,collect_agent_code
,collect_agent_name
,franchisee_code
,franchisee_name
,center_scansitecode
,center_inputsite
,collect_scansitecode
,collect_inputsite
,dispatch_network_code
,dispatch_network_name
,order_source
,collect_planned_dep_date
,collect_planned_departure_time
,center_planned_dep_date
,center_planned_departure_time
,center_planned_arr_date
,center_planned_arrival_time
,center_scantime
,collect_arrival_scan_time
,arrival_scan_inputsite
,send_scan_time
,send_scan_inputsite
,scantime2
,scantime3
,is_dispatch_network
,is_entrepot_exception
,collect_send_in_time
,is_already_flag
,is_ready_flag
,if(is_already_flag=1 or is_ready_flag=1, 0, is_wrong_flag) as is_wrong_flag   --已散货和待散的优先级高，这边错分置为0
,if(collect_send_in_time=0 and is_dispatch_network=0 and is_entrepot_exception=0 and is_collect_arrival=0,1,0) as is_collect_unintime_unarrival --散货不及时且未到件,1标识散货不及时和未到件
,is_collect_arrival
,t_union_2.send_date as update_time
,if(tab_is_reback.waybill_no is not null ,1,0) is_reback
,t_union_2.dt as dt
from t_union_2
left join tab_is_reback on tab_is_reback.waybill_no=t_union_2.billcode
)
insert overwrite table jms_dm.dm_collect_ready_send_detail partition (dt)
select * from t_final_res
distribute by dt,abs(hash(billcode))%9;